﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using TUtils.BasicHelper.Logging;
using TUtils.Multithreading;
using MMessaging.MessagingEx.Common;
using TUtils.BasicHelper;
using System.Threading;
using TUtils.Threads;

namespace MMessaging.MessagingEx.Services
{
	/// <summary>
	/// Repräsentiert einen MessageQueue-Verbinder, der Messages von einer oder mehreren Queues
	/// an eine oder mehrere andere Queues weiterleitet. Er verwendet hierzu ein 
	/// einfaches Broadcast-Verfahren. Die ausgehenden Queues erhalten alle dieselben Nachrichten.
	/// </summary>
	/// <remarks>
	/// 
	/// </remarks>
	/// 
	/// <typeparam name="MESSAGETYPE"></typeparam>
	public class MServiceHubBroadcast<MESSAGETYPE> : TThreadX, IMService<MESSAGETYPE>
		where MESSAGETYPE : class
	{
		#region private 

		private IList<IMChannelEndOut<MESSAGETYPE>> _inputs;
		private IList<IMChannelEndIn<MESSAGETYPE>> _outputs;
		private List<MServiceHubBroadcast<MESSAGETYPE>> _allHubServices = new List<MServiceHubBroadcast<MESSAGETYPE>>();
		/// <summary>
		/// true, wenn der Thread eines anderen Hubs diesen hub mitausführt.
		/// </summary>
		private bool _isZombi = false;

		#region cache
		private List<IMChannelEndOut<MESSAGETYPE>> _allInputs;

		private IList<IMChannelEndOut<MESSAGETYPE>> AllInputs
		{
			get
			{
				if ( _allInputs == null )
				{
					_allInputs = new List<IMChannelEndOut<MESSAGETYPE>>();
					foreach (var hubService in _allHubServices)
					{
						_allInputs.AddRange(hubService._inputs);
					}
				}

				return _allInputs;
			}
		}

		private WaitHandle[] _allWaitHandles;

		private WaitHandle[] AllWaitHandles
		{
			get
			{
				if (_allWaitHandles == null)
				{
					var allWaitHandles = new List<WaitHandle>();
					foreach (var input in AllInputs)
						allWaitHandles.AddRange(input.WaitSignals);

					_allWaitHandles = allWaitHandles.ToArray();
				}

				return _allWaitHandles;
			}
		}

		#endregion

		/// <summary>
		/// 
		/// </summary>
		/// <param name="inputQueue"></param>
		/// <param name="bFoundMessage"></param>
		/// <returns>
		/// false, if Thread should be canceled
		/// true, otherwise
		/// </returns>
		private bool ProcessInputQueue(
			IMChannelEndOut<MESSAGETYPE> inputQueue,
			int inputQueueIdx,
			ref bool bFoundMessage)
		{
			var msg = inputQueue.WaitForMessage(0);
			if (IsCanceling)
				return false;
			if (msg != null)
			{
				bFoundMessage = true;
				var hubService = GetHubService(inputQueueIdx);
				if (hubService != null)
				{
					foreach (var outputQueue in hubService._outputs)
						outputQueue.PostMessage(msg);
				}
			}
			return true;
		}

		private MServiceHubBroadcast<MESSAGETYPE> GetHubService(int inputQueueIdx)
		{
			int curInputQueueIdx = -1;
			foreach (var hubService in _allHubServices)
			{
				curInputQueueIdx += hubService._inputs.Count;
				if (curInputQueueIdx >= inputQueueIdx)
					return hubService;
			}
			return null;
		}


		private void ThreadMethod(IThreadContext context)
		{
			while (!IsCanceling)
			{
				bool bFoundMessage = false;

				var allInputQueues = AllInputs;
				// loop of all input queues
				for(int curInputQueueIdx = 0; curInputQueueIdx < allInputQueues.Count; curInputQueueIdx++)
				{
					// check, if input queue has a message and if yes delegate it.
					// Don't wait for message, just check it
					if (!ProcessInputQueue(
						allInputQueues[curInputQueueIdx],
						curInputQueueIdx,
						ref bFoundMessage))
					{
						return;
					}
				}

				// if no queue has a message
				if (!bFoundMessage)
				{
					// wait at maximum 200 ms for any message in all queues 
					if (AllWaitHandles.Length > 0)
					{
						int inputQueueIdx = WaitHandle.WaitAny(
							AllWaitHandles,
							MMessagingHelper.CANCEL_WAIT_INTERVALL);
						// if a message received
						if (inputQueueIdx != WaitHandle.WaitTimeout)
						{
							if (!ProcessInputQueue(
								AllInputs[inputQueueIdx],
								inputQueueIdx,
								ref bFoundMessage))
							{
								return;
							}
						}
					}
				}
			}
		}

		private void OnSignal(IThreadContext context, SignalEnum signal, Exception e)
		{

		}

		#endregion

		#region constructor

		/// <summary>
		/// Initialisiert den Queue-Verbinder.
		/// </summary>
		/// <param name="name">
		/// Name des Threads, der im Logging erscheinen soll
		/// </param>
		/// <param name="inputs">
		/// Eingangsqueues, die die Messages liefern
		/// 255 at maximum !
		/// </param>
		/// <param name="outputs">
		/// Ausgangsqueues, an die die Messages weitergeleitet werden sollen.
		/// </param>
		/// <param name="startAtOnce">
		/// true, wenn StartAsync() automatisch aufgerufen werden soll.
		/// </param>
		/// <returns></returns>
		public MServiceHubBroadcast<MESSAGETYPE> Init(
			string name,
			IEnumerable<IMChannelEndOut<MESSAGETYPE>> inputs,
			IEnumerable<IMChannelEndIn<MESSAGETYPE>> outputs,
			bool startAtOnce)
		{
			_inputs = new List<IMChannelEndOut<MESSAGETYPE>>(inputs);
			_outputs = new List<IMChannelEndIn<MESSAGETYPE>>(outputs);

			base.Init(
				OnSignal,
				ThreadMethod,
				name);

			_allHubServices.Add(this);

			if (startAtOnce)
				StartAsync();

			return this;
		}

		#endregion

		#region public

		/// <summary>
		/// Verbindet diesen Hub mit dem gegebenen in dearart, dass sie den gleichen
		/// Thread verwenden. darf mehrmals aufgerufen werden.
		/// Verbindet auch dritte bereits mit einem der beiden Hubs verbundene Hubs.
		/// </summary>
		/// <param name="hubService"></param>
		public void Join(MServiceHubBroadcast<MESSAGETYPE> hubService)
		{
			var thisSetOfServices = _allHubServices;
			var otherSetOfServices = hubService._allHubServices;
			var masterOfThisSet = thisSetOfServices.First(service => !service._isZombi);
			var masterOfOtherSet = otherSetOfServices.First(service => !service._isZombi);

			if (masterOfThisSet == masterOfOtherSet)
				// nothing to do
				return;

			if (masterOfOtherSet.Running || masterOfThisSet.Running)
				throw new ApplicationException("hjsadjf324734kdbq634 you may not call Join on running services !");

			masterOfThisSet._allHubServices.AddRange(otherSetOfServices);
			masterOfOtherSet._isZombi = true;
		}

		/// <summary>
		/// Nicht aufrufen, wenn der Hub schon läuft.
		/// </summary>
		/// <param name="channel"></param>
		public void AddToInputs(IMChannelEndOut<MESSAGETYPE> channel)
		{
			if (this.Running)
				throw new ApplicationException("67sehfb238rg allready started.");

			_inputs.Add(channel);
		}

		/// <summary>
		/// Nicht aufrufen, wenn der Hub schon läuft.
		/// </summary>
		/// <param name="channel"></param>
		public void AddToOutputs(IMChannelEndIn<MESSAGETYPE> channel)
		{
			if (this.Running)
				throw new ApplicationException("hx6721fdjaxvqt allready started.");

			_outputs.Add(channel);
		}

		/// <summary>
		/// Nicht aufrufen, wenn der Hub schon läuft.
		/// </summary>
		/// <param name="channel"></param>
		public void RemoveFromInputs(IMChannelEndOut<MESSAGETYPE> channel)
		{
			if (this.Running)
				throw new ApplicationException("hs715ghanaxcvajd allready started.");

			_inputs.Remove(channel);
		}

		/// <summary>
		/// Nicht aufrufen, wenn der Hub schon läuft.
		/// </summary>
		/// <param name="channel"></param>
		public void RemoveFromOutputs(IMChannelEndIn<MESSAGETYPE> channel)
		{
			if (this.Running)
				throw new ApplicationException("g6sdsl9djb2 allready started.");

			_outputs.Remove(channel);
		}

		/// <summary>
		/// Startet Hub asynchron.
		/// Nochmaliger Aufruf eines schon gestarteten Hubs wird ohne Fehler ignoriert.
		/// </summary>
		public override void StartAsync()
		{
			if ( !_isZombi )
				base.StartAsync();
		}

		public string ServiceName
		{
			get { return this.ThreadName; }
		}
		#endregion

	}
}
